package SparkMaster_Worker

import java.util.UUID

import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

import scala.concurrent.duration.DurationInt

object Worker {
  def main(args: Array[String]): Unit = {
    val (workerHost, workerPort, masterHost, masterPort) = ("127.0.0.1", 10006, "127.0.0.1", 10005)
    val conf = ConfigFactory.parseString(
      s"""|akka.actor.provider="akka.remote.RemoteActorRefProvider"
          |akka.remote.netty.tcp.hostname=$workerHost
          |akka.remote.netty.tcp.port=$workerPort
      """.stripMargin
    )
    val workerSystem = ActorSystem("sparkWorker", conf)
    val sparkWorker01 = workerSystem.actorOf(Props(new Worker(masterHost, masterPort)), "sparkWorker01")
    sparkWorker01 ! "start"
  }

}

class Worker(masterHost: String, masterPort: Int) extends Actor {
  //  master 的代理
  var masterPorxy: ActorSelection = _
  val id = UUID.randomUUID().toString

  override def preStart(): Unit = {
    masterPorxy = context.actorSelection(s"akka.tcp://sparkMaster@${masterHost}:${masterPort}/user/sparkMaster01")
  }

  override def receive: Receive = {
    case "start" => {
      println("worker 启动中...")
      masterPorxy ! RegisterWorkerInfo(id, 16, 16 * 1024)
    }
    case RegisterWorkerInfo => {
      println("worker在master中注册成功");
      println("woker 启动成功")
      //定时发送心跳信息
      import context.dispatcher
      context.system.scheduler.schedule(0 millis, 3000 millis, self, SendhertBeat)
    }
    case SendhertBeat => {
      println("发送了一次心跳信息")
      masterPorxy ! HeartBeat(id)
    }
  }
}